package org.myorg.v6;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Timer;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.ipc.RPC;
import org.apache.hadoop.ipc.RPC.Server;

public class ComMonitor implements RegisterProtocol {

	private static final Log LOG = LogFactory.getLog(ComMonitor.class);
	
	private byte[] lock = new byte[0];

	private Server server;

	public Hashtable<String, CommInfo> net = new Hashtable<String, CommInfo>();

	public class CommInfo {

		public Hashtable<InetSocketAddress, Long> nodes = new Hashtable<InetSocketAddress, Long>();

		public CommInfo(String IP, int port) {
			nodes.put(new InetSocketAddress(IP, port), System.currentTimeMillis());
		}
		
		public void put(String IP, int port) {
			nodes.put(new InetSocketAddress(IP, port), System.currentTimeMillis());
		}
	}

	public ComMonitor(){
		try {
			server = RPC.getServer(this, InetAddress.getLocalHost().getHostAddress(), 36241, new Configuration());
			server.start();
			server.join();
		} catch (UnknownHostException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

	@Override
	public  InetSocketAddressWriatble[] register(String ID, String IP, int port, long period) {

		InetSocketAddressWriatble[] rt = new InetSocketAddressWriatble[1];
		synchronized(lock) {

			CommInfo info = (CommInfo) net.get(ID);

			if (info != null) {

				info.put(IP, port); // put the slave IP in list
				
				LinkedList<InetSocketAddress> ips = new LinkedList<InetSocketAddress>();
				
				Enumeration<InetSocketAddress> e = info.nodes.keys();
				for (int i = 0; e.hasMoreElements(); i++) {
					InetSocketAddress st = e.nextElement();
					if (System.currentTimeMillis() - info.nodes.get(st) < period) {
						ips.addLast(st);
					} else {
						info.nodes.remove(ips);
					}
				}
				
				LOG.info("ips.size is " + ips.size());
				rt = new InetSocketAddressWriatble[ips.size()];

				for (int i=0; i < ips.size(); i++) {
					InetSocketAddress st = ips.get(i);
					rt[i] = new InetSocketAddressWriatble(st.getAddress().getHostAddress(), st.getPort());
					LOG.info("return address" + st);
				}

			} else {// info is not exist, this is the first value;
				LOG.info("first comming  " + IP);
				net.put(ID, new CommInfo(IP, port));
				rt[0] = new InetSocketAddressWriatble(IP, port);
			}
		}

		return rt;
	}
	
	
	
	public void clear() {
		Enumeration<String> e = net.keys();
		for (int i = 0; e.hasMoreElements(); i++) {
			String st = e.nextElement();
			if (net.get(st).nodes.isEmpty()) {
				net.remove(st);
			}
		}
	}

	public static void main(String[] args) {
		final ComMonitor m = new ComMonitor();
		Timer timer = new Timer(true); 
		
		timer.schedule(
				new java.util.TimerTask() { 
					public void run() {
						m.clear();
					} 
				}, 
				0, 
				60*1000
		);
		
	}

	@Override
	public long getProtocolVersion(String protocol, long clientVersion)
	throws IOException {
		// TODO Auto-generated method stub
		return 1;
	}



}

